package com.shujia.flink.source;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;

public class Demo04MyMySQLSource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Students> studentDS = env.addSource(new MyMySQLSource());


        // 统计班级人数
        DataStream<Tuple2<String, Integer>> clazzCntDS = studentDS
                .map(stu -> Tuple2.of(stu.clazz, 1), Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(t2 -> t2.f0)
                .sum(1);
        clazzCntDS.print();

        // 统计性别人数
        DataStream<Tuple2<String, Integer>> genderCntDS = studentDS
                .map(stu -> Tuple2.of(stu.gender, 1), Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(t2 -> t2.f0)
                .sum(1);
        genderCntDS.print();

        env.execute();

    }
}

class MyMySQLSource implements SourceFunction<Students> {

    @Override
    public void run(SourceContext<Students> ctx) throws Exception {
        // run方法只会执行一次
        // 建立连接
        Connection conn = DriverManager.getConnection("jdbc:mysql://master:3306/bigdata30", "root", "123456");
        // 创建Statement
        Statement st = conn.createStatement();
        // 执行查询
        ResultSet rs = st.executeQuery("select * from students");
        // 遍历rs提取每一条数据
        while (rs.next()) {
            long id = rs.getLong("id");
            String name = rs.getString("name");
            int age = rs.getInt("age");
            String gender = rs.getString("gender");
            String clazz = rs.getString("clazz");

            Students stu = new Students(id, name, age, gender, clazz);
            ctx.collect(stu);
        }

        rs.close();
        st.close();
        conn.close();

    }

    @Override
    public void cancel() {

    }
}

class Students {
    Long id;
    String name;
    Integer age;
    String gender;
    String clazz;

    public Students(Long id, String name, Integer age, String gender, String clazz) {
        this.id = id;
        this.name = name;
        this.age = age;
        this.gender = gender;
        this.clazz = clazz;
    }
}
